package org.jeecg.modules.kafka;

import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.jeecg.common.util.RedisUtil;
import org.jeecg.modules.basicLock.service.IBasicLockService;
import org.jeecg.modules.lockGps.entity.LockGps;
import org.jeecg.modules.lockGps.service.ILockGpsService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

import java.math.BigDecimal;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;


@Component
@Slf4j
public class KafkaDeviceSendConsumer {

    @Autowired
    private IBasicLockService basicLockService;
    @Autowired
    private RedisUtil redisUtil;
    @Autowired
    private ILockGpsService lockGpsService;

    private final AtomicBoolean listening = new AtomicBoolean(false);
    private final Map<String, Long> latestGpsTimes = new ConcurrentHashMap<>();

    /**
     * 消费者配置，kafka.topic为监听的topic，kafka.group为消费分组，可在yml中修改
     */
    @KafkaListener(topics = "ig2apiDeviceInfo_13Sk5deg7JH55ZmY_CWxK153mic4aJhijVXoxg1UsFF1NeaSc")
    @Async
    public void kafkaListener(ConsumerRecord<String, String> consumerRecord) {
        String value = consumerRecord.value();
        log.info("收到kafka消息ig2apiDeviceInfo：{}", value);
    }

    @KafkaListener(topics = "ig2apiDataInfo_13Sk5deg7JH55ZmY_CWxK153mic4aJhijVXoxg1UsFF1NeaSc")
    @Async
    public void kafkaListenerDeviceData(ConsumerRecord<String, String> consumerRecord) {
        if (!listening.get()) {// 如果没有开始监听，则直接返回
            log.debug("监听成功,直接返回");
            return;
        }
        try {
            String value = consumerRecord.value();
            JSONObject jsonObject = JSONObject.parseObject(value);
            long gpsTime = jsonObject.getLongValue("gpsTime");
            String deviceCode = jsonObject.getString("deviceCode");
            // 从缓存中获取最新GPS时间
            Long latestGpsTime = latestGpsTimes.computeIfAbsent(deviceCode, k -> System.currentTimeMillis());
            log.info("最新GPS时间：{}", latestGpsTime);
            if (gpsTime <= latestGpsTime) {// 检查时间戳是否大于最新的时间戳
                return;// 忽略这条数据，因为它早于阈值
            }
            latestGpsTimes.put(deviceCode, gpsTime);// 更新缓存中的最新GPS时间
            JSONObject data = jsonObject.getJSONObject("data");
            if (data == null) {
                log.warn("数据字段为空");
                return;
            }
            String lng = data.getString("lng");
            String lat = data.getString("lat");
            String carLedgerId = (String) redisUtil.get(deviceCode);
            if (carLedgerId != null) {
                LockGps lockGps = new LockGps();
                lockGps.setDeviceCode(deviceCode).setCarLedgerId(carLedgerId)
                        .setGpsTime(gpsTime).setLat(new BigDecimal(lat)).setLng(new BigDecimal(lng));
                CompletableFuture.runAsync(() -> lockGpsService.save(lockGps));// 异步保存，避免阻塞主线程
            }
        } catch (Exception e) {
            log.error("kafkaListenerDeviceData异常", e);
        }
    }

    public void startListening() {
        listening.set(true);
    }

    public void stopListening() {
        List<String> lockCode = basicLockService.getLockCode();
        boolean lockCodUse = lockCode.stream()
                .anyMatch(key -> redisUtil.hasKey(key));
        if (!lockCodUse){
            listening.set(false);
        }
    }

}
